import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}


object UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SparkStreaming").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    //设置检查点目录
    ssc.checkpoint("/checkpoint")

    ssc.socketTextStream("usdp-o3tbdsfp-monitor1", 4444)
      .flatMap(_.split(" "))
      .map((_, 1))
      .updateStateByKey(updateFunc)
      .print()

    ssc.start()
    ssc.awaitTermination()
  }

  //seq:当前批次单词次数，state:以往批次单词次数
  def updateFunc = (seq: Seq[Int], state: Option[Int]) => {
    //获取当前批次单词的和
    val currentCount: Int = seq.sum
    //获取历史状态的数据
    val stateCount: Int = state.getOrElse(0)
    //将当前批次的和加上历史状态的数据和,返回
    Some(currentCount + stateCount) //some extends Option
  }
}
